package org.apache.rocketmq.client.java.message;

import apache.rocketmq.v2.Resource;
import apache.rocketmq.v2.SystemProperties;
import com.google.protobuf.ByteString;
import com.google.protobuf.util.Timestamps;
import org.apache.rocketmq.client.java.clientapis.message.Message;
import org.apache.rocketmq.client.java.clientapis.message.MessageId;
import org.apache.rocketmq.client.java.impl.producer.ProducerSettings;
import org.apache.rocketmq.client.java.message.protocol.Encoding;
import org.apache.rocketmq.client.java.misc.Utilities;

import java.io.IOException;
import java.util.Optional;

/**
 * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
 * @author：陈清风扬，个人微信号：chenqingfengyangjj。
 * @date:2025/6/16
 * @方法描述：封装要发布的消息的实现类，所谓要发布的消息其实就是要发送给proxy服务端的消息对象
 */
public class PublishingMessageImpl extends MessageImpl {
    //本条消息的Id，也就是消息的唯一标识符
    private final MessageId messageId;

    //要发布的消息的类型，在第十一版本代码中，我们发送的都是普通消息
    private final MessageType messageType;

    //消息跟踪上下文，在源码中这个上下文并没有被用到
    private volatile String traceContext;

    //构造方法
    public PublishingMessageImpl(Message message, ProducerSettings producerSettings, boolean txEnabled) throws IOException {
        super(message);
        this.traceContext = null;
        //得到消息长度
        final int length = message.getBody().remaining();
        //从生产者客户端设置信息中得到允许发布的最大消息的大小
        final int maxBodySizeBytes = producerSettings.getMaxBodySizeBytes();
        //判断本次发送消息是否超过了限制，超出限制则抛出异常
        if (length > maxBodySizeBytes) {
            throw new IOException("Message body size exceeds the threshold, max size=" + maxBodySizeBytes + " bytes");
        }
        //得到消息的唯一Id
        this.messageId = MessageIdCodec.getInstance().nextMessageId();

        //判断当前消息是不是普通消息，如果是普通消息，那么消息是不会进行分组，且不会进行延迟投递，txEnabled表示是否为事务消息
        if (!message.getMessageGroup().isPresent() &&
                !message.getDeliveryTimestamp().isPresent() && !txEnabled) {
            //设置消息类型为普通消息
            messageType = MessageType.NORMAL;
            return;
        }

        //判断当前消息是不是顺序消息，如果消息被分组了，并且不是事务消息，那么该消息就是顺序消息
        if (message.getMessageGroup().isPresent() && !txEnabled) {
            //设置消息类型为顺序消息
            messageType = MessageType.FIFO;
            return;
        }

        //判断当前消息是否为延迟消息
        if (message.getDeliveryTimestamp().isPresent() && !txEnabled) {
            //设置消息类型为延迟消息
            messageType = MessageType.DELAY;
            return;
        }

        //判断当前消息是否为事务消息
        if (!message.getMessageGroup().isPresent() &&
                !message.getDeliveryTimestamp().isPresent() && txEnabled) {
            //设置消息类型为事务消息
            messageType = MessageType.TRANSACTION;
            return;
        }

        //消息类型不明确，则直接抛出异常
        throw new IllegalArgumentException("Transactional message should not set messageGroup or deliveryTimestamp");
    }

    public MessageId getMessageId() {
        return messageId;
    }

    public MessageType getMessageType() {
        return messageType;
    }

    public void setTraceContext(String traceContext) {
        this.traceContext = traceContext;
    }

    public Optional<String> getTraceContext() {
        return null == traceContext ? Optional.empty() : Optional.of(traceContext);
    }


    public apache.rocketmq.v2.Message toProtobuf() {
        final apache.rocketmq.v2.SystemProperties.Builder systemPropertiesBuilder =
                apache.rocketmq.v2.SystemProperties.newBuilder()
                        .addAllKeys(keys)
                        .setMessageId(messageId.toString())
                        .setBornTimestamp(Timestamps.fromMillis(System.currentTimeMillis()))
                        .setBornHost(Utilities.hostName())
                        .setBodyEncoding(Encoding.toProtobuf(Encoding.IDENTITY))
                        .setMessageType(MessageType.toProtobuf(messageType));
        this.getTag().ifPresent(systemPropertiesBuilder::setTag);
        this.getTraceContext().ifPresent(systemPropertiesBuilder::setTraceContext);
        this.getDeliveryTimestamp().ifPresent(millis -> systemPropertiesBuilder.setDeliveryTimestamp(Timestamps.fromMillis(millis)));
        this.getMessageGroup().ifPresent(systemPropertiesBuilder::setMessageGroup);
        final SystemProperties systemProperties = systemPropertiesBuilder.build();
        Resource topicResource = Resource.newBuilder().setName(getTopic()).build();
        return apache.rocketmq.v2.Message.newBuilder()
                .setTopic(topicResource)
                .setBody(ByteString.copyFrom(getBody()))
                .setSystemProperties(systemProperties)
                .putAllUserProperties(getProperties())
                .build();
    }
}
